package com.aotain.nyx.apptype;


import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.util.IPFilterSingleUtil;

public class AppTypeDAllStatisImport implements FlatMapFunction<Tuple2<String,AppTypeTuple>	, Object> {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 6706009868030895225L;
	
	private HashMap<String,String> appTypeMap = new HashMap<String,String>();
	private HashSet<String> appTypeSet = new HashSet<String>();
	public AppTypeDAllStatisImport(String sql ,String dbJson) {
		 appTypeSet = IPFilterSingleUtil.getInstance3(sql,dbJson).AppTypeSet;
		 for(String str:appTypeSet) {
			 String [] splits = str.split("#");
			 String key = splits[0] + "#" + splits[1];
			 appTypeMap.put(key, str);
		 }
		 Logger.getLogger(getClass()).info(String.format("######AppTypeDAllStatisImport### mapSize = %d" ,appTypeMap.size()));
	}

	@Override
	public void flatMap(Tuple2<String, AppTypeTuple> value,
			Collector<Object> out) throws Exception {
		// TODO Auto-generated method stub
		String tableName = "METIS_APP_STAT_IP_D";
		AppTypeTuple col = value.f1;
		if(col == null)
			return;
		String appType = col.getAppType();
		String appId = col.getAppId();
		String appTypeName = "";
		String appName = "";
		String key = appId + "#" + appType;
		String typeValue = appTypeMap.get(key);
		if(typeValue != null) {
			String [] splits = typeValue.split("#",-1);
			appName = splits[2];
			appTypeName = splits[3];
		}
		if("".equals(appName) || "".equals(appTypeName)) {
			return ;
		}
//		Logger.getLogger(getClass()).info(String.format("######AppTypeDStatisImport### appType = %s ; appId = %s ;appName = %s,appTypeName=%s",appType,appId,appName,appTypeName));
		List<DataColumn> saverow = new ArrayList<DataColumn>();
		DataColumn colAppType = new DataColumn();
		colAppType.setKey("APPTYPE");
		colAppType.setValue(appType);
		saverow.add(colAppType);
		DataColumn colAppId = new DataColumn();
		colAppId.setKey("APPID");
		colAppId.setValue(appId);
		saverow.add(colAppId);
		DataColumn colAppName = new DataColumn();
		colAppName.setKey("APPNAME");
		colAppName.setValue(appName);
		saverow.add(colAppName);
		DataColumn colAppTypeName = new DataColumn();
		colAppTypeName.setKey("APPTYPENAME");
		colAppTypeName.setValue(appTypeName);
		saverow.add(colAppTypeName);

		SimpleDateFormat sdfd = new SimpleDateFormat("yyyyMMdd");
		Date date = new Date();
		String dateDay = sdfd.format(date);
		String reporttime = dateDay;

		DataColumn colReportTime = new DataColumn();
		colReportTime.setKey("REPORTTIME");
		colReportTime.setValue(Long.parseLong(reporttime));
		saverow.add(colReportTime);

		List<DataColumn> incrow = new ArrayList<DataColumn>();
		DataColumn colUpStreamOctets = new DataColumn();
		colUpStreamOctets.setKey("UPSTREAMOCTETS");
		colUpStreamOctets.setValue(col.getUpStreamOctets());
		incrow.add(colUpStreamOctets);
		
		DataColumn colUpStreamPacket = new DataColumn();
		colUpStreamPacket.setKey("UPSTREAMPACKET");
		colUpStreamPacket.setValue(col.getUpStreamPacket());
		incrow.add(colUpStreamPacket);
		
		DataColumn colDnStreamOctets = new DataColumn();
		colDnStreamOctets.setKey("DNSTREAMOCTETS");
		colDnStreamOctets.setValue(col.getDnStreamOctets());
		incrow.add(colDnStreamOctets);
		
		DataColumn colDnStreamPacket = new DataColumn();
		colDnStreamPacket.setKey("DNSTREAMPACKET");
		colDnStreamPacket.setValue(col.getDnStreamPacket());
		incrow.add(colDnStreamPacket);

		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("REPORTTIME",Long.parseLong(reporttime)));
		filter.add(new DataColumn("APPTYPE",appType));
		filter.add(new DataColumn("APPID",appId));


		MongoImportTool.getInstance().UpdateRowData(tableName, filter, saverow);

		MongoImportTool.getInstance().IncRowData(tableName, filter, incrow);
	}

}
